package com.hskn.hss.config;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.hskn.hss.module.kfkanalyse.service.IkfkAnalyseService;
import com.hskn.hss.module.kfkbadge.entity.KfkBadge;
import com.hskn.hss.module.kfkbadge.mapper.KfkbadgeMapper;
import com.hskn.hss.module.kfkcarvideo.mapper.KfkCarVideoMapper;
import com.hskn.hss.module.kfkdeal.service.IkfkDealService;
import com.hskn.hss.module.kfkdeal.to.KfkDealEmployeeTo;
import com.hskn.hss.module.kfkdtu.mapper.KfkDtuMapper;
import com.hskn.hss.module.kfkhumancamer.mapper.KfkHumanCamerMapper;
import com.hskn.hss.module.kfkinterphone.mapper.KfkInterphoneMapper;
import com.hskn.hss.module.kfknvr.mapper.KfkNvrMapper;
import com.hskn.hss.module.kfkpunchcard.mapper.KfkPunchCardMapper;
import com.hskn.hss.module.kfkworkcard.mapper.KfkWorkCardMapper;
import com.hskn.hss.module.tlemployee.entity.TlEmployee;
import com.hskn.hss.module.tlemployee.mapper.TlEmployeeMapper;
import com.hskn.hss.module.tlequip.entity.TlEquip;
import com.hskn.hss.module.tlequip.mapper.TlEquipMapper;
import com.hskn.hss.utils.KafkaUtils;
import com.hskn.hss.utils.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.Date;
import java.util.Map;

/**
 * @author lny
 * createTime 2021-09-09
 **/
@Order(value = 1)
@Slf4j
@Component
public class KfkBadgeRunner implements CommandLineRunner {

    @Autowired
    private IkfkDealService ikfkDealService;
    @Autowired
    private KfkbadgeMapper kfkbadgeMapper;
    @Autowired
    private TlEquipMapper tlEquipMapper;
    @Autowired
    TlEmployeeMapper tlEmployeeMapper;
    @Autowired
    KfkWorkCardMapper kfkWorkCardMapper;
    @Autowired
    IkfkAnalyseService ikfkAnalyseService;
    @Autowired
    KfkInterphoneMapper kfkInterphoneMapper;
    @Autowired
    KfkPunchCardMapper kfkPunchCardMapper;
    @Autowired
    KfkHumanCamerMapper kfkHumanCamerMapper;
    @Autowired
    KfkDtuMapper kfkDtuMapper;
    @Autowired
    KfkCarVideoMapper kfkCarVideoMapper;
    @Autowired
    KfkNvrMapper kfkNvrMapper;

    @Value("${kafka.badge_topic}")
    private String badge_topic;

    @Value("${kafka.badge_group}")
    private String badge_group;

    @Value("${kafka.kafka_receive:true}")
    private Boolean kafka_receive;

    @Override
    public void run(String... args) {
        if (!kafka_receive) {
            return;
        }
        new Thread(() -> {
            KafkaConsumer consumer = KafkaUtils.getConsumer(badge_group);
            consumer.subscribe(Arrays.asList(badge_topic));
            while (true) {
                receiveAndHandle(consumer);
            }
        }).start();
    }

    public void receiveAndHandle(KafkaConsumer consumer) {
        //拉取kafka数据
        log.info("====================kfk定时读取任务开始-电子校徽======================================");
        try {
            ConsumerRecords<Object, Object> records = consumer.poll(1000);
            log.info("========KfkBadgeRunner拉取条数：{}", records.count());
            for (ConsumerRecord<Object, Object> record : records) {
                try {
                    Map value = JSONObject.parseObject(String.valueOf(record.value()), Map.class);
                    Map content = JSONObject.parseObject(String.valueOf(value.get("content")), Map.class);
                    Map info = JSONObject.parseObject(String.valueOf(content.get("info")), Map.class);
                    Date timestamp = new Date(record.timestamp());

                    if (value.get("deviceType").equals("WORK_BADGE") && content.get("type").equals("GPS_LOCATION_REPORT")) { // 电子校徽
                        long workBadgeStartTime = System.currentTimeMillis();
                        Date deviceTime = StringUtils.StringToDate(String.valueOf(info.get("loctime")));
                        KfkBadge kfkbadge = new KfkBadge();
                        kfkbadge.setEqpnum(String.valueOf(value.get("no")));
                        kfkbadge.setState(String.valueOf(value.get("state")));
                        kfkbadge.setLat(String.valueOf(info.get("la")));
                        kfkbadge.setLon(String.valueOf(info.get("lo")));
                        kfkbadge.setDeviceTime(deviceTime);//设备时间
                        kfkbadge.setSpeed(0d);
                        kfkbadge.setDirection(0d);
                        kfkbadge.setTimestamp(timestamp);//kafka时间
                        kfkbadge.setBat(String.valueOf(info.get("bat")));
                        kfkbadge.setGsm(String.valueOf(info.get("gsm")));
                        kfkbadge.setSatelliteNum(Integer.parseInt(info.get("gps").toString()));
                        kfkbadge.setLocationtype(String.valueOf(info.get("locationtype")));
                        kfkbadge.setIsonline(info.get("isonline") == null ? null : String.valueOf(info.get("isonline")));
                        kfkbadge.setSecond(0);
                        kfkbadge.setSos(info.get("sos") == null ? null : String.valueOf(info.get("sos")));
                        kfkbadge.setCreateTime(new Date());//创建时间
                        kfkbadgeMapper.insert(kfkbadge);

                        QueryWrapper<TlEquip> queryWrapper = new QueryWrapper<>();
                        queryWrapper.lambda().eq(TlEquip::getEquipcode, kfkbadge.getEqpnum());
                        TlEquip tlEquip = tlEquipMapper.selectOne(queryWrapper);
                        if (null != tlEquip) {
                            QueryWrapper<TlEmployee> queryWrapper2 = new QueryWrapper<>();
                            queryWrapper2.lambda().eq(TlEmployee::getId, tlEquip.getBindobjid());
                            TlEmployee tlEmployee = tlEmployeeMapper.selectOne(queryWrapper2);
                            if (null != tlEmployee) {
                                tlEmployee.setAddr(kfkbadge.getLon() + "," + kfkbadge.getLat());
                                tlEmployeeMapper.updateById(tlEmployee);
                            }
                        }
                        KfkDealEmployeeTo kfkDealEmployeeTo = new KfkDealEmployeeTo();
                        kfkDealEmployeeTo.setEqp_type(3);
                        kfkDealEmployeeTo.setDate(deviceTime);
                        kfkDealEmployeeTo.setLng(kfkbadge.getLon());
                        kfkDealEmployeeTo.setLat(kfkbadge.getLat());
                        kfkDealEmployeeTo.setEquanum(kfkbadge.getEqpnum());
                        kfkDealEmployeeTo.setEmployee_id(tlEquip.getBindobjid());
                        kfkDealEmployeeTo.setSpeed(kfkbadge.getSpeed().intValue());
                        kfkDealEmployeeTo.setDirection(kfkbadge.getDirection().intValue());
                        ikfkDealService.KfkDealEmployee(kfkDealEmployeeTo);
                        long workBadgeEndTime = System.currentTimeMillis();
                        log.info("=====电子校徽=执行时间(毫秒)：{}", workBadgeEndTime - workBadgeStartTime);
                    }
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
            log.info("====================kfk定时读取任务结束-电子校徽======================================");
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        } finally {
            consumer.commitAsync();
        }
    }
}